package com.cmit.upay.monitor.monitor;

import com.cmit.upay.monitor.properties.ActiveMQProperties;
import com.cmit.upay.monitor.service.WriteDataService;
import com.cmit.upay.monitor.util.DateUtils;
import com.cmit.upay.monitor.util.JsonUtil;
import com.cmit.upay.monitor.vo.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.QueueViewMBean;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import javax.management.MBeanServerConnection;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;

/**
 * 监控activemq
 */
@Slf4j
@Component
public class ActiveMQMonitor {
    private static final String JMX_SERVICE_URL_PREFIX = "service:jmx:rmi:///jndi/rmi://";
    private static final String JMX_SERVICE_URL_SUFFIX = "/jmxrmi";

    private static final String JOLOKIA_URL_QUEUES_PATTERN = "http://%s/api/jolokia/read/org.apache.activemq:brokerName=%s,type=Broker/Queues";
    private static final String JOLOKIA_QUEUE_URL_PATTERN = "http://%s/api/jolokia/read/org.apache.activemq:brokerName=%s,type=Broker,destinationType=Queue,destinationName=%s";
    private static final String CONSUMERS_URL = "http://%s/admin/queueConsumers.jsp?JMSDestination=%s";


    @Autowired
    private ActiveMQProperties activeMQProperties;

    @Autowired
    private Map<String, WriteDataService> serviceMap;

    private Map<String, JMXConnector> connectorMap = new HashMap<>();

    @Autowired
    private RestTemplate restTemplate;

    /**
     * 请求队列数据
     * @param adminUrl
     */
    @Async
    public void monitorUseJolokia(String adminUrl) {
        List<String> queuesList = getJolokiaQueueList(adminUrl);
        log.info("获取到{}共{}个队列。", adminUrl, queuesList.size());
        List<QueueVO> dataList = getJolokiaData(queuesList, adminUrl);
        log.info("获取{}的队列数据完成", adminUrl);
        WriteDataService service = serviceMap.get(adminUrl);
        service.write(dataList);
        log.info("{}写入数据完成", adminUrl);
    }

    private List<String> getJolokiaQueueList(String adminUrl) {
        String queuesUrl = String.format(JOLOKIA_URL_QUEUES_PATTERN, adminUrl, activeMQProperties.getBorkerName());
        log.info("请求地址{}", queuesUrl);
        List<String> queuesList = getQueuesList(queuesUrl);
        // 如果有指定队列
        if (activeMQProperties.getQueuesList().size() > 0) {
            queuesList = getMonitorList(queuesList);
        }
        if (log.isDebugEnabled()) {
            log.debug("获取{}所有队列名{}成功", adminUrl, queuesList);
        }
        return queuesList;
    }

    /**
     * 请求消费者里的信息<br/>
     * 对应网页中{ip}:{port}/admin/queueConsumers.jsp?JMSDestination=xxx的信息
     */
    @Async
    public void requestConsumersInfo(String adminUrl) {
        List<String> queuesList = getJolokiaQueueList(adminUrl);
        log.info("获取到{}共{}个队列。", adminUrl, queuesList.size());
        requestConsumerInfo(adminUrl, queuesList);
    }

    /**
     * 获取队列的消费者信息
     * @param queuesList 队列名列表
     * @param adminUrl
     */
    public void requestConsumerInfo(String adminUrl, List<String> queuesList) {
        for (String queueName : queuesList) {
            requestConsumerInfo(adminUrl, queueName);
            log.info("获取队列{}中consumer信息完成", queueName);
        }
    }

    /**
     * 获取队列的消费者信息
     * @param queueName 队列名列表
     * @param adminUrl
     */
    public void requestConsumerInfo(String adminUrl, String queueName) {
        String requestUrl = String.format(CONSUMERS_URL, adminUrl, queueName);
        List<ConsumerInfo> consumerInfoList = consumerInfoList(requestUrl);
        writeToFile(adminUrl, queueName, consumerInfoList);
        log.info("队列{}中的consumer信息写入文件完成", queueName);
    }

    /**
     * 请求队列中的所有consumers信息
     * @param url
     * @return
     */
    public List<ConsumerInfo> consumerInfoList(String url) {
        List<ConsumerInfo> result = new ArrayList<>();
        // 请求队列对应的Consumer信息
        String response = restTemplate.exchange(url, HttpMethod.GET, getAuthEntity(activeMQProperties), String.class).getBody();
        if (log.isDebugEnabled()) {
            log.debug("收到mq返回数据{}", response);
        }
        // 转换实体类
        Document document = Jsoup.parse(response);
        Elements rows = document.getElementById("messages").select("tr");
        // 从index = 1开始迭代，因为第0行是表头信息
        // TODO 这一段比较难以维护，element到bean的转换需要重构
        for (int i = 1; i < rows.size(); i++) {
            Element element = rows.get(i);
            Elements cols = element.select("td");
            ConsumerInfo.ConsumerInfoBuilder builder = ConsumerInfo.builder();
            ConsumerInfo consumerInfo = builder.monitorTime(DateUtils.getCurrentTime())
                    .clientId(cols.get(0).text())
                    .sessionId(cols.get(1).text())
                    .enqueues(Long.parseLong(cols.get(3).text()))
                    .dequeues(Long.parseLong(cols.get(4).text()))
                    .dispatched(Long.parseLong(cols.get(5).text()))
                    .dispatchedQueue(cols.get(6).text())
                    .prefetch(Long.parseLong(cols.get(7).text().split(" ")[0]))
                    .maxPending(Long.parseLong(cols.get(7).text().split(" ")[1]))
                    .build();
            result.add(consumerInfo);
            if (log.isDebugEnabled()) {
                log.debug("解析到consumer信息{}", consumerInfo);
            }
        }
        return result;
    }

    public void writeToFile(String adminUrl, String queueName, List<ConsumerInfo> consumerInfoList) {
        String filename = queueName;
        Path outputFileDir = Paths.get(activeMQProperties.getOutputDir())
                .resolve(adminUrl.replaceAll(":", "_"))
                .resolve("consumers");
        File currentFile = getCurrentFile(outputFileDir, filename);
        filename = currentFile.getName();// 更新一下filename
        try {
            // 追加模式写文件
            BufferedWriter fileWriter = new BufferedWriter(new FileWriter(currentFile, true));
            for (ConsumerInfo info : consumerInfoList) {
                fileWriter.write(info.toString());
                fileWriter.newLine();
            }
            // 所有数据写完后flush一下
            fileWriter.flush();
        } catch (IOException e) {
            log.error(String.format("写队列consumer文件过程出错", queueName), e);
        }

    }

    /**
     * 文件名采用：queueName.0, queueName.1, queueName.2的方式叠加
     * @param outputFileDir
     * @param filename
     * @return
     */
    public File getCurrentFile(Path outputFileDir, String filename) {
        try {
            Files.createDirectories(outputFileDir);
            File[] files = outputFileDir.toFile().listFiles();
            int max = 0;// 记录当前写入的最大文件
            for (File file : files) {
                // 过滤出当前队列名对应的文件
                if (!file.getName().startsWith(filename)) {
                    continue;
                }
                int index = file.getName().lastIndexOf(".");
                String fileIndex = file.getName().substring(index + 1);
                if (Integer.parseInt(fileIndex) > max) {
                    max = Integer.parseInt(fileIndex);
                }
            }
            // 当前获取到的最后一个写入的文件
            String newFilename = filename + "." + max;
            File currentFile = Paths.get(outputFileDir.toString()).resolve(newFilename).toFile();
            if (currentFile.length() > activeMQProperties.getConsumerFileSize()) {// 大于10M，写新文件
                max++;
                newFilename = filename + "." + max;
                currentFile = Paths.get(outputFileDir.toString()).resolve(newFilename).toFile();
            }
            log.info("当前写入文件{}", newFilename);
            return currentFile;
        } catch (Exception e) {
            log.error("获取当前需要写的文件名出错", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 使用JMX监控
     * @param jmxUrl
     * @deprecated 需要重启MQ和配置JMX，直接通过JOLOKIA接口更方便
     */
    @Deprecated
    @Async
    public void monitorUseJmx(String jmxUrl) {
        String url = JMX_SERVICE_URL_PREFIX + jmxUrl + JMX_SERVICE_URL_SUFFIX;
        MBeanServerConnection connection = getConnection(url);
        log.info("获取{}连接完成", jmxUrl);
        List<QueueVO> dataList = getData(connection);
        log.info("获取{}的队列数据完成", jmxUrl);
        WriteDataService service = serviceMap.get(jmxUrl);
        service.write(dataList);
        log.info("{}写入数据完成", jmxUrl);
        closeConnector(jmxUrl);
        log.info("关闭jmx={}连接", jmxUrl);
    }

    /**
     * 获取mq中的队列数据
     *
     * @param connection
     * @return
     */
    public List<QueueVO> getData(MBeanServerConnection connection) {
        List<QueueVO> list = new ArrayList<>();

        try {
            ObjectName objectName = new ObjectName(String.format("org.apache.activemq:brokerName=%s,type=Broker", activeMQProperties.getBorkerName()));
            BrokerViewMBean mBean = MBeanServerInvocationHandler.newProxyInstance(connection,
                    objectName, BrokerViewMBean.class, true);

            for (ObjectName na : mBean.getQueues()) {
                QueueVO.QueueVOBuilder builder = QueueVO.builder();
                // 获取队列的MBean
                QueueViewMBean queueBean = MBeanServerInvocationHandler.newProxyInstance(connection,
                        na, QueueViewMBean.class, true);
                // 有指定监控队列
                if (activeMQProperties.getQueuesList().size() > 0) {
                    if (!activeMQProperties.getQueuesList().contains(queueBean.getName())) {
                        continue;
                    }
                }

                // 构造vo对象
                builder.monitorTime(DateUtils.getCurrentTime())
                        .name(queueBean.getName())
                        .queueSize(queueBean.getQueueSize())
                        .consumerCount(queueBean.getConsumerCount())
                        .producerCount(queueBean.getProducerCount())
                        .enqueueCount(queueBean.getEnqueueCount())
                        .dequeueCount(queueBean.getDequeueCount());

                list.add(builder.build());
            }
        } catch (Exception e) {
            String errorMsg = "获取队列相关数据出现异常";
            log.error(errorMsg);
            throw new RuntimeException(errorMsg, e);
        } finally {

        }
        return list;
    }


    /**
     * 获取jmx连接
     *
     * @param jmxUrl
     * @return
     */
    public MBeanServerConnection getConnection(String jmxUrl) {
        JMXServiceURL url = null;
        JMXConnector connector = null;
        try {
            url = new JMXServiceURL(jmxUrl);
            connector = JMXConnectorFactory.connect(url);
            connector.connect();
            connectorMap.put(jmxUrl, connector);
            return connector.getMBeanServerConnection();
        } catch (Exception e) {
            String errorMsg = String.format("获取jmx连接%s出现异常出现异常", jmxUrl);
            log.error(errorMsg);
            throw new RuntimeException(errorMsg, e);
        }
    }

    private void closeConnector(String jmxUrl) {
        // 移除
        JMXConnector connector = connectorMap.remove(jmxUrl);
        if (connector != null) {
            try {
                connector.close();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    public List<String> getQueuesList(String queuesUrl) {
        List<String> result = new ArrayList<>();
        // 请求所有队列数据
        String response = restTemplate.exchange(queuesUrl, HttpMethod.GET, getAuthEntity(activeMQProperties), String.class).getBody();
        if (log.isDebugEnabled()) {
            log.debug("收到mq返回数据{}", response);
        }
        // 转换成实体类
        JolokiaQueueList queueList = JsonUtil.jsonStr2Obj(response, JolokiaQueueList.class);
        // 获取队列名
        for (QueueListValue value : queueList.getValue()) {
            result.add(value.getDestinationName());
        }
        return result;
    }

    public List<QueueVO> getJolokiaData(List<String> queuesList, String adminUrl) {
        List<QueueVO> result = new ArrayList<>();

        // 逐个队列请求
        for (String queueName : queuesList) {
            // 请求url
            String queueUrl = String.format(JOLOKIA_QUEUE_URL_PATTERN, adminUrl, activeMQProperties.getBorkerName(), queueName);
            // 发送请求
            String response = restTemplate.exchange(queueUrl, HttpMethod.GET, getAuthEntity(activeMQProperties), String.class).getBody();
            if (log.isDebugEnabled()) {
                log.debug("收到mq返回数据{}", response);
            }
            QueueVO queueVO = jsonToQueueVo(response);
            result.add(queueVO);
        }
        return result;
    }

    /**
     * value 形如：
     * {"request":{"mbean":"org.apache.activemq:brokerName=localhost,destinationName=test1,destinationType=Queue,type=Broker","type":"read"},"value":{"ProducerFlowControl":true,"Options":"","AlwaysRetroactive":false,"MemoryUsageByteCount":2116,"AverageBlockedTime":0.0,"MemoryPercentUsage":0,"CursorMemoryUsage":2116,"InFlightCount":0,"Subscriptions":[],"CacheEnabled":true,"ForwardCount":0,"DLQ":false,"StoreMessageSize":670,"AverageEnqueueTime":0.0,"Name":"test1","TotalBlockedTime":0,"MaxAuditDepth":2048,"BlockedSends":0,"QueueSize":2,"MaxPageSize":200,"TempUsageLimit":27942535168,"PrioritizedMessages":false,"MemoryUsagePortion":1.0,"Paused":false,"EnqueueCount":2,"MessageGroups":{},"TempUsagePercentUsage":0,"ConsumerCount":0,"AverageMessageSize":1058,"CursorFull":false,"MaxProducersToAudit":1024,"ExpiredCount":0,"CursorPercentUsage":0,"MinEnqueueTime":0,"MemoryLimit":668309914,"MinMessageSize":1058,"DispatchCount":0,"MaxEnqueueTime":0,"DequeueCount":0,"BlockedProducerWarningInterval":30000,"ProducerCount":0,"MessageGroupType":"cached","MaxMessageSize":1058,"UseCache":true,"SlowConsumerStrategy":null},"timestamp":1583919313,"status":200}
     * @param value
     * @return
     */
    private QueueVO jsonToQueueVo(String value) {
        QueueVO.QueueVOBuilder builder = QueueVO.builder();

        // json转换成实体类
        JolokiaQueueMBean queueMBean = JsonUtil.jsonStr2Obj(value, JolokiaQueueMBean.class);

        // 构造QueueVo
        QueueVO vo = QueueVO.builder()
                .monitorTime(DateUtils.getCurrentTime())
                .build();
        // 将mbean的属性复制到vo里
        BeanUtils.copyProperties(queueMBean.getValue(), vo);

        return vo;
    }

    public List<String> getMonitorList(List<String> queuesList) {
        List<String> result = new ArrayList<>();
        for (String queue : queuesList) {
            if (activeMQProperties.getQueuesList().contains(queue)) {
                result.add(queue);
            }
        }
        return result;
    }

    /**
     * 获取请求实体，包含basic认证
     * @param properties
     * @return
     */
    private HttpEntity<String> getAuthEntity(ActiveMQProperties properties) {
        // 拼接用户名密码
        String userMsg = properties.getAdminUsername() + ":" + properties.getAdminPassword();
        // base64编码
        String base64UserMsg = Base64.getEncoder().encodeToString(userMsg.getBytes());
        // 构造请求http头
        HttpHeaders headers = new HttpHeaders();
        headers.add("Authorization", "Basic " + base64UserMsg);
        // 构造entity
        HttpEntity<String> entity = new HttpEntity<>(headers);
        return entity;
    }

    public void setActiveMQProperties(ActiveMQProperties activeMQProperties) {
        this.activeMQProperties = activeMQProperties;
    }

    public void setRestTemplate(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }
}
